package com.shujia.streaming

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SparkSession}

object Code07StructStreaming {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[2]")
      .appName("StructuredNetworkWordCount")
      .getOrCreate()

    import spark.implicits._

    val dataFrame: DataFrame = spark
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", 8888)
//      .schema("line String")
      .load()

    import org.apache.spark.sql.functions._
    dataFrame
      .toDF("line")
//      .select(explode(split($"line", ",")) as "word")
//      .groupBy("word")
//      .agg(count("*") as "cnt")
//      .show()  不能直接使用show
      .writeStream  // 以流的形式将数据写出

      /**
       * OutputMode模式：
       *  1.Complete 保存历史批次数据，并且只能用于带有聚合操作的DF
       *  2.Append 对当前批次数据处理，不能带有聚合操作
       *  3.Update 可以包含聚合或者不包含聚合 并且只对当前批次数据有效
       */
//      .outputMode(OutputMode.Complete())
      .outputMode(OutputMode.Update())
      .format("console") // 写出到控制台
      // 和SparkStreamingContext写法一致
      .start()
      .awaitTermination()


    // 该方式会对之前的历史数据进行保存
  }
}
